package gupiaodown;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.stream.io.ByteArrayOutputStream;

import net.minidev.json.JSONObject;
import net.minidev.json.JSONValue;

@Tags({" download gupiao price "})
@CapabilityDescription("gupiao price download ")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class downLoadguPriceProcessor extends AbstractProcessor {

	public downLoadguPriceProcessor() {
		// TODO Auto-generated constructor stub
	}

	public static final Relationship MY_RELATIONSHIPOK = new Relationship.Builder()
            .name("gen success")
            .description("success relationship")
            .build();
	public static final Relationship MY_RELATIONSHIPNO = new Relationship.Builder()
            .name("gen failed")
            .description("failed relationship")
            .build();
	
	private List<PropertyDescriptor> descriptors;
    private Set<Relationship> relationships;
    
    
    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

 @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
       
        this.descriptors = Collections.unmodifiableList(descriptors);
        final Set<Relationship> relationships = new HashSet<Relationship>();
        relationships.add(MY_RELATIONSHIPOK);
        relationships.add(MY_RELATIONSHIPNO);
        this.relationships = Collections.unmodifiableSet(relationships);
    }
 
	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		// TODO Auto-generated method stub
		
		FlowFile flowFile = session.get();
	      if ( flowFile == null ) {
	      	
	      	System.out.println("flowfile is null");
	         return;
	      }
	      
	      //读取FlowFile
	        String content = "";
	        InputStream in = session.read(flowFile);
	        try {
				byte[] by = readStream(in);
				content = new String(by,"UTF-8");
				
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				//return ;
			}
	        //content的内容就是url  code name
	        DownAndAnalyzeGupiaoFromSina dotask = new DownAndAnalyzeGupiaoFromSina();
	        
	        JSONObject obj = (JSONObject) JSONValue.parse(content);
	        String url = (String) obj.get("url");
	        String code =  (String) obj.get("code");
	        String name =  (String) obj.get("name");
	        
	        String page = dotask.downLoadOnePage(url);
	        String strresult  = dotask.analyzePage(url, page,code,name);
	        //
	        flowFile= session.write(flowFile, new OutputStreamCallback(){
	    		@Override
	    		public void process(OutputStream out) throws IOException{
	    			out.write(strresult.getBytes("UTF-8"));
	    		}
	    	}); 
	        session.transfer(flowFile,MY_RELATIONSHIPOK);
		
	}
	
	
	public static byte[] readStream(InputStream inStream) throws Exception {  
        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
        byte[] buffer = new byte[1024];  
        int len = -1;  
        while ((len = inStream.read(buffer)) != -1) {  
            outSteam.write(buffer, 0, len);  
        }  
        outSteam.close();  
        inStream.close();  
        return outSteam.toByteArray();  
    }  

}
